package cn.dmrliu.edu.realtime.util;

import cn.dmrliu.edu.realtime.bean.TransientSink;
import cn.dmrliu.edu.realtime.common.EduConfig;
import org.apache.flink.connector.jdbc.*;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class ClickhouseUtil {

    public static <T> SinkFunction<T> getSinkFunction(String sql) {

        SinkFunction<T> sink = JdbcSink.<T>sink(
                sql,
                new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement ps, T obj) throws SQLException {
                        // 使用反射为对象赋值
                        Field[] fieldArr = obj.getClass().getDeclaredFields();
                        int skipNum = 0;
                        for (int i = 0; i < fieldArr.length; i++) {
                            Field field = fieldArr[i];
                            // 如果有跳过注解，跳过赋值
                            TransientSink annotation = field.getAnnotation(TransientSink.class);
                            if (annotation != null) {
                                skipNum++;
                                continue;
                            }
                            // 为对象赋值
                            field.setAccessible(true);
                            try {
                                Object fieldValue = field.get(obj);
                                ps.setObject(i + 1 - skipNum, fieldValue);
                            } catch (IllegalAccessException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                },
                new JdbcExecutionOptions.Builder()
                        .withBatchSize(5)
                        .withBatchIntervalMs(3000L)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(EduConfig.CLICKHOUSE_DRIVER)
                        .withUrl(EduConfig.CLICKHOUSE_URL)
                        .withUsername(EduConfig.CLICKHOUSE_USERNAME)
                        .withPassword(EduConfig.CLICKHOUSE_PASSWORD)
                        .build()
        );

        return sink;
    }
}
